/*
 * #%L
 * Alfresco Repository
 * %%
 * Copyright (C) 2005 - 2023 Alfresco Software Limited
 * %%
 * This file is part of the Alfresco software. 
 * If the software was purchased under a paid Alfresco license, the terms of 
 * the paid license agreement will prevail.  Otherwise, the software is 
 * provided under the following open source license terms:
 * 
 * Alfresco is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * Alfresco is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with Alfresco. If not, see <http://www.gnu.org/licenses/>.
 * #L%
 */
package org.alfresco.repo.transfer;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Serializable;
import java.io.Writer;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.parsers.SAXParser;
import javax.xml.parsers.SAXParserFactory;
import jakarta.transaction.UserTransaction;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.xml.sax.SAXException;

import org.alfresco.error.AlfrescoRuntimeException;
import org.alfresco.model.ContentModel;
import org.alfresco.repo.cache.SimpleCache;
import org.alfresco.repo.transaction.RetryingTransactionHelper.RetryingTransactionCallback;
import org.alfresco.repo.transfer.manifest.TransferManifestDeletedNode;
import org.alfresco.repo.transfer.manifest.TransferManifestHeader;
import org.alfresco.repo.transfer.manifest.TransferManifestNode;
import org.alfresco.repo.transfer.manifest.TransferManifestNodeFactory;
import org.alfresco.repo.transfer.manifest.TransferManifestNodeHelper;
import org.alfresco.repo.transfer.manifest.TransferManifestNormalNode;
import org.alfresco.repo.transfer.manifest.TransferManifestProcessor;
import org.alfresco.repo.transfer.manifest.TransferManifestWriter;
import org.alfresco.repo.transfer.manifest.XMLTransferManifestReader;
import org.alfresco.repo.transfer.manifest.XMLTransferManifestWriter;
import org.alfresco.repo.transfer.report.TransferReporter;
import org.alfresco.repo.transfer.requisite.DeltaListRequsiteProcessor;
import org.alfresco.repo.transfer.requisite.XMLTransferRequsiteReader;
import org.alfresco.service.cmr.action.Action;
import org.alfresco.service.cmr.action.ActionService;
import org.alfresco.service.cmr.repository.ChildAssociationRef;
import org.alfresco.service.cmr.repository.ContentData;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.service.cmr.search.SearchService;
import org.alfresco.service.cmr.transfer.TransferCallback;
import org.alfresco.service.cmr.transfer.TransferCancelledException;
import org.alfresco.service.cmr.transfer.TransferDefinition;
import org.alfresco.service.cmr.transfer.TransferEndEvent;
import org.alfresco.service.cmr.transfer.TransferEvent;
import org.alfresco.service.cmr.transfer.TransferEventCancelled;
import org.alfresco.service.cmr.transfer.TransferEventError;
import org.alfresco.service.cmr.transfer.TransferEventReport;
import org.alfresco.service.cmr.transfer.TransferEventSuccess;
import org.alfresco.service.cmr.transfer.TransferException;
import org.alfresco.service.cmr.transfer.TransferFailureException;
import org.alfresco.service.cmr.transfer.TransferProgress;
import org.alfresco.service.cmr.transfer.TransferService2;
import org.alfresco.service.cmr.transfer.TransferTarget;
import org.alfresco.service.cmr.transfer.TransferVersion;
import org.alfresco.service.descriptor.Descriptor;
import org.alfresco.service.descriptor.DescriptorService;
import org.alfresco.service.namespace.NamespaceService;
import org.alfresco.service.namespace.QName;
import org.alfresco.service.namespace.RegexQNamePattern;
import org.alfresco.service.transaction.TransactionService;
import org.alfresco.util.PropertyCheck;
import org.alfresco.util.TempFileProvider;

/**
 * Implementation of the Transfer Service.
 * 
 * @author Mark Rogers
 *
 */
public class TransferServiceImpl2 implements TransferService2
{
    private static Log logger = LogFactory.getLog(TransferServiceImpl2.class);

    private static final String MSG_NO_HOME = "transfer_service.unable_to_find_transfer_home";
    private static final String MSG_NO_GROUP = "transfer_service.unable_to_find_transfer_group";
    private static final String MSG_NO_TARGET = "transfer_service.unable_to_find_transfer_target";
    private static final String MSG_ERR_TRANSFER_ASYNC = "transfer_service.unable_to_transfer_async";
    private static final String MSG_TARGET_EXISTS = "transfer_service.target_exists";
    private static final String MSG_NO_NODES = "transfer_service.no_nodes";
    private static final String MSG_MISSING_ENDPOINT_PATH = "transfer_service.missing_endpoint_path";
    private static final String MSG_MISSING_ENDPOINT_PROTOCOL = "transfer_service.missing_endpoint_protocol";
    private static final String MSG_MISSING_ENDPOINT_HOST = "transfer_service.missing_endpoint_host";
    private static final String MSG_MISSING_ENDPOINT_PORT = "transfer_service.missing_endpoint_port";
    private static final String MSG_MISSING_ENDPOINT_USERNAME = "transfer_service.missing_endpoint_username";
    private static final String MSG_MISSING_ENDPOINT_PASSWORD = "transfer_service.missing_endpoint_password";
    private static final String MSG_FAILED_TO_GET_TRANSFER_STATUS = "transfer_service.failed_to_get_transfer_status";
    private static final String MSG_TARGET_ERROR = "transfer_service.target_error";
    private static final String MSG_UNKNOWN_TARGET_ERROR = "transfer_service.unknown_target_error";
    private static final String MSG_TARGET_NOT_ENABLED = "transfer_service.target_not_enabled";
    private static final String MSG_INCOMPATIBLE_VERSIONS = "transfer_service.incompatible_versions";

    private static final String FILE_DIRECTORY = "transfer";
    private static final String FILE_SUFFIX = ".xml";

    private enum ClientTransferState
    {
        Begin, Prepare, Commit, Poll, Cancel, Finished, Exit;
    };

    /**
     * The synchronised list of transfers in progress.
     */
    private Map<String, TransferStatus> transferMonitoring = Collections.synchronizedMap(new TreeMap<String, TransferStatus>());

    public void init()
    {
        PropertyCheck.mandatory(this, "nodeService", nodeService);
        PropertyCheck.mandatory(this, "searchService", searchService);
        PropertyCheck.mandatory(this, "transferSpaceQuery", transferSpaceQuery);
        PropertyCheck.mandatory(this, "defaultTransferGroup", defaultTransferGroup);
        PropertyCheck.mandatory(this, "transmitter", transmitter);
        PropertyCheck.mandatory(this, "namespaceResolver", transmitter);
        PropertyCheck.mandatory(this, "actionService", actionService);
        PropertyCheck.mandatory(this, "transactionService", transactionService);
        PropertyCheck.mandatory(this, "descriptorService", descriptorService);
        PropertyCheck.mandatory(this, "transferVersionChecker", transferVersionChecker);
    }

    private String transferSpaceQuery;
    private String defaultTransferGroup;
    private NodeService nodeService;
    private SearchService searchService;
    private TransferTransmitter transmitter;
    private TransactionService transactionService;
    private ActionService actionService;
    private TransferManifestNodeFactory transferManifestNodeFactory;
    private TransferReporter transferReporter;
    private DescriptorService descriptorService;
    private TransferVersionChecker transferVersionChecker;
    private NamespaceService namespaceService;

    /**
     * How long to delay while polling for commit status.
     */
    private long commitPollDelay = 2000;

    /**
     * Create a new in memory transfer target
     */
    public TransferTarget createTransferTarget(String name)
    {
        NodeRef dummy = lookupTransferTarget(name);
        if (dummy != null)
        {
            throw new TransferException(MSG_TARGET_EXISTS, new Object[]{name});
        }

        TransferTargetImpl newTarget = new TransferTargetImpl();
        newTarget.setName(name);
        return newTarget;
    }

    /**
     * create transfer target
     */
    public TransferTarget createAndSaveTransferTarget(String name, String title, String description, String endpointProtocol, String endpointHost, int endpointPort, String endpointPath, String username, char[] password)
    {
        TransferTargetImpl newTarget = new TransferTargetImpl();
        newTarget.setName(name);
        newTarget.setTitle(title);
        newTarget.setDescription(description);
        newTarget.setEndpointProtocol(endpointProtocol);
        newTarget.setEndpointHost(endpointHost);
        newTarget.setEndpointPort(endpointPort);
        newTarget.setEndpointPath(endpointPath);
        newTarget.setUsername(username);
        newTarget.setPassword(password);
        return createTransferTarget(newTarget);

    }

    /**
     * create transfer target
     */
    private TransferTarget createTransferTarget(TransferTarget newTarget)
    {
        /**
         * Check whether name is already used
         */
        NodeRef dummy = lookupTransferTarget(newTarget.getName());
        if (dummy != null)
        {
            throw new TransferException(MSG_TARGET_EXISTS,
                    new Object[]{newTarget.getName()});
        }

        Map<QName, Serializable> properties = new HashMap<QName, Serializable>();

        // type properties
        properties.put(TransferModel.PROP_ENDPOINT_HOST, newTarget.getEndpointHost());
        properties.put(TransferModel.PROP_ENDPOINT_PORT, newTarget.getEndpointPort());
        properties.put(TransferModel.PROP_ENDPOINT_PROTOCOL, newTarget.getEndpointProtocol());
        properties.put(TransferModel.PROP_ENDPOINT_PATH, newTarget.getEndpointPath());
        properties.put(TransferModel.PROP_USERNAME, newTarget.getUsername());
        properties.put(TransferModel.PROP_PASSWORD, new String(encrypt(newTarget.getPassword())));

        // titled aspect
        properties.put(ContentModel.PROP_TITLE, newTarget.getTitle());
        properties.put(ContentModel.PROP_NAME, newTarget.getName());
        properties.put(ContentModel.PROP_DESCRIPTION, newTarget.getDescription());

        // enableable aspect
        properties.put(TransferModel.PROP_ENABLED, Boolean.TRUE);

        NodeRef defaultGroup = getDefaultGroup();

        /**
         * Go ahead and create the new node
         */
        ChildAssociationRef ref = nodeService.createNode(defaultGroup, ContentModel.ASSOC_CONTAINS,
                QName.createQName(TransferModel.TRANSFER_MODEL_1_0_URI, newTarget.getName()),
                TransferModel.TYPE_TRANSFER_TARGET, properties);

        /**
         * Now create a new TransferTarget object to return to the caller.
         */
        TransferTargetImpl retVal = new TransferTargetImpl();
        mapTransferTarget(ref.getChildRef(), retVal);

        return retVal;
    }

    protected NodeRef getDefaultGroup()
    {
        NodeRef home = getTransferHome();
        List<ChildAssociationRef> refs = nodeService.getChildAssocs(home, ContentModel.ASSOC_CONTAINS, QName.createQName(defaultTransferGroup, namespaceService));
        if (refs.isEmpty())
        {
            // No transfer group.
            throw new TransferException(MSG_NO_GROUP, new Object[]{
                    defaultTransferGroup
            });
        }
        return refs.get(0).getChildRef();
    }

    /**
     * Get all transfer targets
     */
    public Set<TransferTarget> getTransferTargets()
    {
        NodeRef home = getTransferHome();

        Set<TransferTarget> ret = new HashSet<TransferTarget>();

        // get all groups
        List<ChildAssociationRef> groups = nodeService.getChildAssocs(home);

        // for each group
        for (ChildAssociationRef group : groups)
        {
            NodeRef groupNode = group.getChildRef();
            ret.addAll(getTransferTargets(groupNode));
        }

        return ret;
    }

    /**
     * Get all transfer targets in the specified group
     */
    public Set<TransferTarget> getTransferTargets(String groupName)
    {
        NodeRef home = getTransferHome();

        // get group with assoc groupName
        NodeRef groupNode = nodeService.getChildByName(home, ContentModel.ASSOC_CONTAINS, groupName);

        if (groupNode == null)
        {
            // No transfer group.
            throw new TransferException(MSG_NO_GROUP, new Object[]{groupName});
        }

        return getTransferTargets(groupNode);
    }

    /**
     * Given the noderef of a group of transfer targets, return all the contained transfer targets.
     * 
     * @param groupNode
     *            NodeRef
     * @return Set<TransferTarget>
     */
    private Set<TransferTarget> getTransferTargets(NodeRef groupNode)
    {
        Set<TransferTarget> result = new HashSet<TransferTarget>();
        List<ChildAssociationRef> children = nodeService.getChildAssocs(groupNode, ContentModel.ASSOC_CONTAINS, RegexQNamePattern.MATCH_ALL);

        for (ChildAssociationRef child : children)
        {
            if (nodeService.getType(child.getChildRef()).equals(TransferModel.TYPE_TRANSFER_TARGET))
            {
                TransferTargetImpl newTarget = new TransferTargetImpl();
                mapTransferTarget(child.getChildRef(), newTarget);
                result.add(newTarget);
            }
        }
        return result;
    }

    /**
     * 
     */
    public void deleteTransferTarget(String name)
    {
        NodeRef nodeRef = lookupTransferTarget(name);

        if (nodeRef == null)
        {
            // target does not exist
            throw new TransferException(MSG_NO_TARGET, new Object[]{name});
        }
        nodeService.deleteNode(nodeRef);
    }

    /**
     * Enables/Disables the named transfer target
     */
    public void enableTransferTarget(String name, boolean enable)
    {
        NodeRef nodeRef = lookupTransferTarget(name);
        nodeService.setProperty(nodeRef, TransferModel.PROP_ENABLED, Boolean.valueOf(enable));
    }

    public boolean targetExists(String name)
    {
        return (lookupTransferTarget(name) != null);
    }

    /**
     * 
     */
    public TransferTarget getTransferTarget(String name)
    {
        NodeRef nodeRef = lookupTransferTarget(name);

        if (nodeRef == null)
        {
            // target does not exist
            throw new TransferException(MSG_NO_TARGET, new Object[]{name});
        }
        TransferTargetImpl newTarget = new TransferTargetImpl();
        mapTransferTarget(nodeRef, newTarget);

        return newTarget;
    }

    /**
     * create or update a transfer target.
     */
    public TransferTarget saveTransferTarget(TransferTarget update)
    {
        if (update.getNodeRef() == null)
        {
            // This is a save for the first time
            return createTransferTarget(update);
        }

        NodeRef nodeRef = lookupTransferTarget(update.getName());
        if (nodeRef == null)
        {
            // target does not exist
            throw new TransferException(MSG_NO_TARGET, new Object[]{update.getName()});
        }

        Map<QName, Serializable> properties = new HashMap<QName, Serializable>();
        properties.put(TransferModel.PROP_ENDPOINT_HOST, update.getEndpointHost());
        properties.put(TransferModel.PROP_ENDPOINT_PORT, update.getEndpointPort());
        properties.put(TransferModel.PROP_ENDPOINT_PROTOCOL, update.getEndpointProtocol());
        properties.put(TransferModel.PROP_ENDPOINT_PATH, update.getEndpointPath());
        properties.put(TransferModel.PROP_USERNAME, update.getUsername());
        properties.put(TransferModel.PROP_PASSWORD, new String(encrypt(update.getPassword())));

        // titled aspect
        properties.put(ContentModel.PROP_TITLE, update.getTitle());
        properties.put(ContentModel.PROP_NAME, update.getName());
        properties.put(ContentModel.PROP_DESCRIPTION, update.getDescription());

        properties.put(TransferModel.PROP_ENABLED, Boolean.valueOf(update.isEnabled()));
        nodeService.setProperties(nodeRef, properties);

        TransferTargetImpl newTarget = new TransferTargetImpl();
        mapTransferTarget(nodeRef, newTarget);
        return newTarget;
    }

    /**
     * Transfer async.
     * 
     * @param targetName
     *            String
     * @param definition
     *            TransferDefinition
     * @param callbacks
     *            TransferCallback...
     * 
     */
    public void transferAsync(String targetName, TransferDefinition definition, TransferCallback... callbacks)
    {
        transferAsync(targetName, definition, Arrays.asList(callbacks));
    }

    /**
     * Transfer async.
     * 
     * @param targetName
     *            String
     * @param definition
     *            TransferDefinition
     * 
     */
    public void transferAsync(String targetName, TransferDefinition definition, Collection<TransferCallback> callbacks)
    {
        /**
         * Event processor for this transfer instance
         */
        final TransferEventProcessor eventProcessor = new TransferEventProcessor();
        if (callbacks != null)
        {
            eventProcessor.observers.addAll(callbacks);
        }

        /* Note: callback should be Serializable to be passed through the action API However Serializable is not used so it does not matter. Perhaps the action API should be changed? Or we could add a Serializable proxy here. */

        Map<String, Serializable> params = new HashMap<String, Serializable>();
        params.put("targetName", targetName);
        params.put("definition", definition);
        params.put("callbacks", (Serializable) callbacks);

        Action transferAction = actionService.createAction("transfer-async", params);

        /**
         * Execute transfer async in its own transaction. The action service only runs actions in the post commit which is why there's a separate transaction here.
         */
        boolean success = false;
        UserTransaction trx = transactionService.getNonPropagatingUserTransaction();
        try
        {
            trx.begin();
            logger.debug("calling action service to execute action");
            actionService.executeAction(transferAction, null, false, true);
            trx.commit();
            logger.debug("committed successfully");
            success = true;
        }
        catch (Exception error)
        {
            logger.error("unexpected exception", error);
            throw new AlfrescoRuntimeException(MSG_ERR_TRANSFER_ASYNC, error);
        }
        finally
        {
            if (!success)
            {
                try
                {
                    logger.debug("rolling back after error");
                    trx.rollback();
                }
                catch (Exception error)
                {
                    logger.error("unexpected exception during rollback", error);
                    // There's nothing much we can do here
                }
            }
        }
    }

    /**
     * Transfer Synchronous
     * 
     * @param targetName
     *            String
     * @param definition
     *            TransferDefinition
     * @param callbacks
     *            TransferCallback...
     */
    public TransferEndEvent transfer(String targetName, TransferDefinition definition, TransferCallback... callbacks)
            throws TransferFailureException
    {
        return transfer(targetName, definition, Arrays.asList(callbacks));
    }

    /**
     * Transfer Synchronous
     * 
     * @param targetName
     *            String
     * @param definition
     *            TransferDefinition
     */
    public TransferEndEvent transfer(String targetName, TransferDefinition definition, Collection<TransferCallback> callbacks)
            throws TransferFailureException
    {
        /**
         * Event processor for this transfer instance
         */
        final TransferEventProcessor eventProcessor = new TransferEventProcessor();
        if (callbacks != null)
        {
            eventProcessor.observers.addAll(callbacks);
        }

        /**
         * Now go ahead and do the transfer
         */
        return transferImpl(targetName, definition, eventProcessor);
    }

    private TransferEndEvent transferImpl(String targetName, final TransferDefinition definition, final TransferEventProcessor eventProcessor)
            throws TransferFailureException
    {
        if (logger.isDebugEnabled())
        {
            logger.debug("transfer started to :" + targetName);
        }

        // transfer end event
        TransferEndEvent endEvent = null;
        Exception failureException = null;
        TransferTarget target = null;
        Transfer transfer = null;
        final List<TransferEvent> transferReportEvents = new LinkedList<TransferEvent>();
        NodeRef sourceReport = null;
        NodeRef destinationReport = null;
        File manifest = null;
        File requisite = null;
        int pollRetries = 0;
        int pollPosition = -1;
        boolean cancelled = false;

        Descriptor currentDescriptor = descriptorService.getCurrentRepositoryDescriptor();
        Descriptor serverDescriptor = descriptorService.getServerDescriptor();
        final String localRepositoryId = currentDescriptor.getId();
        TransferVersion fromVersion = new TransferVersionImpl(serverDescriptor);

        // Wire in the transferReport - so any callbacks are stored in transferReport
        TransferCallback reportCallback = new TransferCallback() {
            public void processEvent(TransferEvent event)
            {
                transferReportEvents.add(event);
            }
        };
        eventProcessor.addObserver(reportCallback);

        TransferContext transferContext = new TransferContext();

        // start transfer
        ClientTransferState clientState = ClientTransferState.Begin;
        while (clientState != ClientTransferState.Exit)
        {
            try
            {
                switch (clientState)
                {
                case Begin:
                {
                    eventProcessor.start();

                    manifest = createManifest(definition, localRepositoryId, fromVersion, transferContext);
                    logger.debug("transfer begin");
                    target = getTransferTarget(targetName);
                    checkTargetEnabled(target);
                    transfer = transmitter.begin(target, localRepositoryId, fromVersion);
                    String transferId = transfer.getTransferId();
                    TransferStatus status = new TransferStatus();
                    transferMonitoring.put(transferId, status);
                    logger.debug("transfer begun transferId:" + transferId);
                    eventProcessor.begin(transferId);
                    checkCancel(transferId);

                    // next state
                    clientState = ClientTransferState.Prepare;
                    break;
                }

                case Prepare:
                {
                    // check alfresco versions are compatible
                    TransferVersion toVersion = transfer.getToVersion();
                    if (!this.transferVersionChecker.checkTransferVersions(fromVersion, toVersion))
                    {
                        throw new TransferException(MSG_INCOMPATIBLE_VERSIONS, new Object[]{transfer.getTransferId(), fromVersion, toVersion});
                    }

                    // send Manifest, get the requsite back.
                    eventProcessor.sendSnapshot(1, 1);

                    requisite = createRequisiteFile();
                    FileOutputStream reqOutput = new FileOutputStream(requisite);
                    transmitter.sendManifest(transfer, manifest, reqOutput);
                    logger.debug("manifest sent");
                    checkCancel(transfer.getTransferId());

                    if (logger.isDebugEnabled())
                    {
                        logger.debug("requisite file written to local filesystem");
                        try
                        {
                            outputFile(requisite);
                        }
                        catch (IOException error)
                        {
                            // This is debug code - so an exception thrown while debugging
                            logger.debug("error while outputting snapshotFile");
                            error.printStackTrace();
                        }
                    }

                    sendContent(transfer, definition, eventProcessor, manifest, requisite);
                    logger.debug("content sending finished");
                    checkCancel(transfer.getTransferId());

                    // prepare
                    eventProcessor.prepare();
                    transmitter.prepare(transfer);
                    checkCancel(transfer.getTransferId());

                    // next state
                    clientState = ClientTransferState.Commit;
                    break;
                }

                case Commit:
                {
                    logger.debug("about to start committing transferId:" + transfer.getTransferId());
                    eventProcessor.commit();
                    transmitter.commit(transfer);

                    logger.debug("committing transferId:" + transfer.getTransferId());
                    checkCancel(transfer.getTransferId());

                    // next state
                    clientState = ClientTransferState.Poll;
                    break;
                }

                case Poll:
                {
                    TransferProgress progress = null;
                    try
                    {
                        progress = transmitter.getStatus(transfer);

                        // reset retries for next poll
                        pollRetries = 0;
                    }
                    catch (TransferException e)
                    {
                        pollRetries++;
                        if (pollRetries == 3)
                        {
                            throw new TransferException(MSG_FAILED_TO_GET_TRANSFER_STATUS, new Object[]{target.getName()});
                        }
                    }

                    // check status
                    if (progress.getStatus() == TransferProgress.Status.ERROR)
                    {
                        Throwable targetError = progress.getError();
                        // NOTE: it's possible the error is not returned from pre v3.4 target repositories
                        if (targetError == null)
                        {
                            targetError = new TransferException(MSG_UNKNOWN_TARGET_ERROR);
                        }
                        if (Exception.class.isAssignableFrom(targetError.getClass()))
                        {
                            failureException = (Exception) targetError;
                        }
                        else
                        {
                            failureException = new TransferException(MSG_TARGET_ERROR, new Object[]{targetError.getMessage()}, targetError);
                        }
                        clientState = ClientTransferState.Finished;
                        break;
                    }
                    else if (progress.getStatus() == TransferProgress.Status.CANCELLED)
                    {
                        cancelled = true;
                        clientState = ClientTransferState.Finished;
                        break;
                    }

                    // notify transfer progress
                    if (progress.getCurrentPosition() != pollPosition)
                    {
                        pollPosition = progress.getCurrentPosition();
                        logger.debug("committing :" + pollPosition);
                        eventProcessor.committing(progress.getEndPosition(), pollPosition);
                    }

                    if (progress.getStatus() == TransferProgress.Status.COMPLETE)
                    {
                        clientState = ClientTransferState.Finished;
                        break;
                    }

                    checkCancel(transfer.getTransferId());

                    // NOTE: stay in poll state...
                    // sleep before next poll
                    try
                    {
                        Thread.sleep(commitPollDelay);
                    }
                    catch (InterruptedException e)
                    {
                        // carry on
                    }
                    break;
                }

                case Cancel:
                {
                    logger.debug("Abort - waiting for target confirmation of cancel");
                    transmitter.abort(transfer);

                    // next state... poll for confirmation of cancel from target
                    clientState = ClientTransferState.Poll;
                    break;
                }

                case Finished:
                {
                    try
                    {
                        TransferEndEventImpl endEventImpl = null;
                        String reportName = null;

                        try
                        {
                            if (failureException != null)
                            {
                                logger.debug("TransferException - unable to transfer", failureException);
                                TransferEventError errorEvent = new TransferEventError();
                                errorEvent.setTransferState(TransferEvent.TransferState.ERROR);
                                errorEvent.setException(failureException);
                                errorEvent.setMessage(failureException.getMessage());
                                endEventImpl = errorEvent;
                                reportName = "error";
                            }
                            else if (cancelled)
                            {
                                endEventImpl = new TransferEventCancelled();
                                endEventImpl.setTransferState(TransferEvent.TransferState.CANCELLED);
                                endEventImpl.setMessage("cancelled");
                                reportName = "cancelled";
                            }
                            else
                            {
                                logger.debug("committed transferId:" + transfer.getTransferId());
                                endEventImpl = new TransferEventSuccess();
                                endEventImpl.setTransferState(TransferEvent.TransferState.SUCCESS);
                                endEventImpl.setMessage("success");
                                reportName = "success";
                            }

                            // manually add the terminal event to the transfer report event list
                            transferReportEvents.add(endEventImpl);
                        }
                        catch (Exception e)
                        {
                            // report this failure as last resort
                            failureException = e;
                            reportName = "error";
                            logger.warn("Exception - unable to notify end transfer state", e);
                        }

                        reportName += "_" + new SimpleDateFormat("yyyyMMddhhmmssSSS").format(new Date());

                        try
                        {
                            if (transfer != null)
                            {
                                logger.debug("now pull back the destination transfer report");
                                destinationReport = persistDestinationTransferReport(reportName, transfer, target);
                                if (destinationReport != null)
                                {
                                    eventProcessor.writeReport(destinationReport, TransferEventReport.ReportType.DESTINATION, endEventImpl.getTransferState());
                                }
                            }

                            logger.debug("now persist the client side transfer report");
                            sourceReport = persistTransferReport(reportName, transfer, target, definition, transferReportEvents, manifest, failureException);
                            if (sourceReport != null)
                            {
                                eventProcessor.writeReport(sourceReport, TransferEventReport.ReportType.SOURCE, endEventImpl.getTransferState());
                            }
                        }
                        catch (Exception e)
                        {
                            logger.warn("Exception - unable to write transfer reports", e);
                        }

                        try
                        {
                            endEventImpl.setLast(true);
                            endEventImpl.setSourceReport(sourceReport);
                            endEventImpl.setDestinationReport(destinationReport);
                            endEvent = endEventImpl;
                            eventProcessor.end(endEvent);
                        }
                        catch (Exception e)
                        {
                            // report this failure as last resort
                            failureException = e;
                            logger.warn("Exception - unable to notify end transfer state", e);
                        }
                    }
                    finally
                    {
                        clientState = ClientTransferState.Exit;
                    }
                }
                }
            }
            catch (TransferCancelledException e)
            {
                logger.debug("Interrupted by transfer cancel request from client");
                clientState = ClientTransferState.Cancel;
            }
            catch (Exception e)
            {
                logger.debug("Exception - unable to transfer", e);

                /**
                 * Save the first exception that we encounter.
                 */
                if (failureException == null)
                {
                    failureException = e;
                }
                if (transfer != null && (clientState == ClientTransferState.Begin ||
                        clientState == ClientTransferState.Prepare ||
                        clientState == ClientTransferState.Commit))
                {
                    // we must first inform the target repository that a client failure has occurred to allow it to
                    // clean up appropriately, too
                    clientState = ClientTransferState.Cancel;
                }
                else
                {
                    clientState = ClientTransferState.Finished;
                }
            }
        }

        try
        {
            if (endEvent == null)
            {
                TransferEventError error = new TransferEventError();
                error.setTransferState(TransferEvent.TransferState.ERROR);
                TransferFailureException endException = new TransferFailureException(error);
                error.setMessage(endException.getMessage());
                error.setException(endException);
                error.setSourceReport(sourceReport);
                error.setDestinationReport(destinationReport);
                error.setLast(true);
                endEvent = error;
            }
            if (endEvent instanceof TransferEventError)
            {
                TransferEventError endError = (TransferEventError) endEvent;
                throw new TransferFailureException(endError);
            }
            return endEvent;
        }
        finally
        {
            // clean up
            if (transfer != null)
            {
                transferMonitoring.remove(transfer.getTransferId());
            }
            if (manifest != null)
            {
                manifest.delete();
                logger.debug("manifest file deleted");
            }

            if (requisite != null)
            {
                requisite.delete();
                logger.debug("requisite file deleted");
            }
        }
    }

    private File createManifest(TransferDefinition definition, String repositoryId, TransferVersion fromVersion, TransferContext transferContext)
            throws IOException, SAXException
    {
        // which nodes to write to the snapshot
        Set<NodeRef> nodes = definition.getNodes();
        Set<NodeRef> nodesToRemove = definition.getNodesToRemove();

        if ((nodes == null || nodes.size() == 0) && (nodesToRemove == null || nodesToRemove.size() == 0))
        {
            logger.debug("no nodes to transfer");
            throw new TransferException(MSG_NO_NODES);
        }

        // If a noderef exists in both the "nodes" set and the "nodesToRemove" set then the nodesToRemove wins. It is removed
        // from the "nodes" set...
        if (nodes != null && nodesToRemove != null)
        {
            nodes.removeAll(nodesToRemove);
        }

        int nodeCount = ((nodes == null) ? 0 : nodes.size()) + ((nodesToRemove == null) ? 0 : nodesToRemove.size());

        /**
         * create snapshot
         */
        logger.debug("create snapshot");

        // where to put snapshot ?
        File tempDir = TempFileProvider.getLongLifeTempDir(FILE_DIRECTORY);
        File snapshotFile = TempFileProvider.createTempFile("TRX-SNAP", FILE_SUFFIX, tempDir);
        Writer snapshotWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(snapshotFile), "UTF-8"));

        // Write the manifest file
        TransferManifestWriter formatter = new XMLTransferManifestWriter();
        TransferManifestHeader header = new TransferManifestHeader();
        header.setRepositoryId(repositoryId);
        header.setTransferVersion(fromVersion);
        header.setCreatedDate(new Date());
        header.setNodeCount(nodeCount);
        header.setSync(definition.isSync());
        header.setReadOnly(definition.isReadOnly());
        formatter.startTransferManifest(snapshotWriter);
        formatter.writeTransferManifestHeader(header);
        if (nodes != null)
        {
            for (NodeRef nodeRef : nodes)
            {
                TransferManifestNode node = transferManifestNodeFactory.createTransferManifestNode(nodeRef, definition, transferContext);
                formatter.writeTransferManifestNode(node);
            }
        }
        if (nodesToRemove != null)
        {
            for (NodeRef nodeRef : nodesToRemove)
            {
                TransferManifestNode node = transferManifestNodeFactory.createTransferManifestNode(nodeRef, definition, transferContext,
                        true);
                formatter.writeTransferManifestNode(node);
            }
        }
        formatter.endTransferManifest();
        snapshotWriter.close();

        logger.debug("snapshot file written to local filesystem");
        // If we are debugging then write the file to stdout.
        if (logger.isDebugEnabled())
        {
            try
            {
                outputFile(snapshotFile);
            }
            catch (IOException error)
            {
                // This is debug code - so an exception thrown while debugging
                logger.debug("error while outputting snapshotFile");
                error.printStackTrace();
            }
        }

        return snapshotFile;
    }

    private File createRequisiteFile()
    {
        File tempDir = TempFileProvider.getLongLifeTempDir(FILE_DIRECTORY);
        File reqFile = TempFileProvider.createTempFile("TRX-REQ", FILE_SUFFIX, tempDir);
        return reqFile;
    }

    private void sendContent(final Transfer transfer, final TransferDefinition definition, final TransferEventProcessor eventProcessor,
            File manifest, File requisite)
            throws SAXException, ParserConfigurationException, IOException
    {
        SAXParserFactory saxParserFactory = SAXParserFactory.newInstance();
        SAXParser parser;
        parser = saxParserFactory.newSAXParser();

        /**
         * Parse the requisite file to generate the delta list
         */
        DeltaListRequsiteProcessor reqProcessor = new DeltaListRequsiteProcessor();
        XMLTransferRequsiteReader reqReader = new XMLTransferRequsiteReader(reqProcessor);
        parser.parse(requisite, reqReader);

        final DeltaList deltaList = reqProcessor.getDeltaList();

        /**
         * Parse the manifest file and transfer chunks over
         * 
         * ManifestFile -> Manifest Processor -> Chunker -> Transmitter
         * 
         * Step 1: Create a chunker and wire it up to the transmitter
         */
        final ContentChunker chunker = new ContentChunkerImpl();
        final Long removeNodesRange = Long.valueOf(definition.getNodesToRemove() != null ? definition.getNodesToRemove().size() : 0);
        final Long nodesRange = Long.valueOf(definition.getNodes() != null ? definition.getNodes().size() : 0);

        final Long fRange = removeNodesRange + nodesRange;
        chunker.setHandler(
                new ContentChunkProcessor() {
                    private long counter = 0;

                    public void processChunk(Set<ContentData> data)
                    {
                        checkCancel(transfer.getTransferId());
                        logger.debug("send chunk to transmitter");
                        for (ContentData file : data)
                        {
                            counter++;
                            eventProcessor.sendContent(file, fRange, counter);
                        }
                        transmitter.sendContent(transfer, data);
                    }
                });

        /**
         * Step 2 : create a manifest processor and wire it up to the chunker
         */
        TransferManifestProcessor processor = new TransferManifestProcessor() {
            public void processTransferManifestNode(TransferManifestNormalNode node)
            {
                Set<ContentData> data = TransferManifestNodeHelper.getContentData(node);
                for (ContentData d : data)
                {
                    checkCancel(transfer.getTransferId());
                    logger.debug("add content to chunker");

                    /**
                     * Check with the deltaList whether we need to send the content item
                     */
                    if (deltaList != null)
                    {
                        String partName = TransferCommons.URLToPartName(d.getContentUrl());
                        if (deltaList.getRequiredParts().contains(partName))
                        {
                            logger.debug("content is required :" + d.getContentUrl());
                            chunker.addContent(d);
                        }
                    }
                    else
                    {
                        // No delta list - so send all content items
                        chunker.addContent(d);
                    }
                }
            }

            public void processTransferManifiestHeader(TransferManifestHeader header)
            {/* NO-OP */ }

            public void startTransferManifest()
            { /* NO-OP */ }

            public void endTransferManifest()
            { /* NO-OP */ }

            public void processTransferManifestNode(TransferManifestDeletedNode node)
            { /* NO-OP */
            }
        };

        /**
         * Step 3: wire up the manifest reader to a manifest processor
         */

        XMLTransferManifestReader reader = new XMLTransferManifestReader(processor);

        /**
         * Step 4: start the magic - Give the manifest file to the manifest reader
         */
        parser.parse(manifest, reader);
        chunker.flush();
    }

    /**
     * CancelAsync
     */
    public void cancelAsync(String transferHandle)
    {
        TransferStatus status = transferMonitoring.get(transferHandle);
        if (status != null)
        {
            logger.debug("canceling transfer :" + transferHandle);
            status.cancelMe = true;
        }
    }

    /**
     * Check whether the specified transfer should be cancelled.
     * 
     * @param transferHandle
     *            String
     * @throws TransferException
     *             - the transfer has been cancelled.
     */
    private void checkCancel(String transferHandle) throws TransferException
    {
        TransferStatus status = transferMonitoring.get(transferHandle);
        if (status != null)
        {
            if (!status.cancelInProgress && status.cancelMe)
            {
                status.cancelInProgress = true;
                throw new TransferCancelledException();
            }
        }
    }

    private void checkTargetEnabled(TransferTarget target) throws TransferException
    {
        if (!target.isEnabled())
        {
            logger.debug("target is not enabled");
            throw new TransferException(MSG_TARGET_NOT_ENABLED, new Object[]{target.getName()});
        }
    }

    public void setNodeService(NodeService nodeService)
    {
        this.nodeService = nodeService;
    }

    public void setSearchService(SearchService searchService)
    {
        this.searchService = searchService;
    }

    public void setSingletonCache(SimpleCache<String, NodeRef> singletonCache)
    {
        this.singletonCache = singletonCache;
    }

    public void setTransferSpaceQuery(String transferSpaceQuery)
    {
        this.transferSpaceQuery = transferSpaceQuery;
    }

    public void setDefaultTransferGroup(String defaultGroup)
    {
        this.defaultTransferGroup = defaultGroup;
    }

    public TransferTransmitter getTransmitter()
    {
        return transmitter;
    }

    public void setTransmitter(TransferTransmitter transmitter)
    {
        this.transmitter = transmitter;
    }

    // note: cache is tenant-aware (if using TransctionalCache impl)

    private SimpleCache<String, NodeRef> singletonCache; // eg. for transferHomeNodeRef
    private final String KEY_TRANSFER_HOME_NODEREF = "key.transferServiceImpl2Home.noderef";

    protected NodeRef getTransferHome()
    {
        NodeRef transferHome = singletonCache.get(KEY_TRANSFER_HOME_NODEREF);
        if (transferHome == null)
        {
            String query = transferSpaceQuery;
            List<NodeRef> refs = searchService.selectNodes(nodeService.getRootNode(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE), query, null, namespaceService, false);
            if (refs.size() == 0)
            {
                // No transfer home.
                throw new TransferException(MSG_NO_HOME, new Object[]{query});
            }
            if (refs.size() != 0)
            {
                transferHome = refs.get(0);
                singletonCache.put(KEY_TRANSFER_HOME_NODEREF, transferHome);
            }
        }
        return transferHome;
    }

    private char[] encrypt(char[] text)
    {
        // placeholder dummy implementation - add an 'E' to the start
        // String dummy = new String("E" + text);
        // String dummy = new String(text);
        // return dummy.toCharArray();
        return text;
    }

    private char[] decrypt(char[] text)
    {
        // placeholder dummy implementation - strips off leading 'E'
        // String dummy = new String(text);
        return text;
        // return dummy.substring(1).toCharArray();
    }

    /**
     * 
     * @param name
     *            String
     * @return NodeRef
     */
    private NodeRef lookupTransferTarget(String name)
    {
        NodeRef defaultGroup = getDefaultGroup();
        return nodeService.getChildByName(defaultGroup, ContentModel.ASSOC_CONTAINS, name);
    }

    private void mapTransferTarget(NodeRef nodeRef, TransferTargetImpl def)
    {
        def.setNodeRef(nodeRef);
        Map<QName, Serializable> properties = nodeService.getProperties(nodeRef);
        String name = (String) properties.get(ContentModel.PROP_NAME);

        String endpointPath = (String) properties.get(TransferModel.PROP_ENDPOINT_PATH);
        if (endpointPath == null)
            throw new TransferException(MSG_MISSING_ENDPOINT_PATH, new Object[]{name});
        def.setEndpointPath(endpointPath);

        String endpointProtocol = (String) properties.get(TransferModel.PROP_ENDPOINT_PROTOCOL);
        if (endpointProtocol == null)
            throw new TransferException(MSG_MISSING_ENDPOINT_PROTOCOL, new Object[]{name});
        def.setEndpointProtocol(endpointProtocol);

        String endpointHost = (String) properties.get(TransferModel.PROP_ENDPOINT_HOST);
        if (endpointHost == null)
            throw new TransferException(MSG_MISSING_ENDPOINT_HOST, new Object[]{name});
        def.setEndpointHost(endpointHost);

        Integer endpointPort = (Integer) properties.get(TransferModel.PROP_ENDPOINT_PORT);
        if (endpointPort == null)
            throw new TransferException(MSG_MISSING_ENDPOINT_PORT, new Object[]{name});
        def.setEndpointPort(endpointPort);

        String username = (String) properties.get(TransferModel.PROP_USERNAME);
        if (username == null)
            throw new TransferException(MSG_MISSING_ENDPOINT_USERNAME, new Object[]{name});
        def.setUsername(username);

        Serializable passwordVal = properties.get(TransferModel.PROP_PASSWORD);
        if (passwordVal == null)
            throw new TransferException(MSG_MISSING_ENDPOINT_PASSWORD, new Object[]{name});
        if (passwordVal.getClass().isArray())
        {
            def.setPassword(decrypt((char[]) passwordVal));
        }
        if (passwordVal instanceof String)
        {
            String password = (String) passwordVal;
            def.setPassword(decrypt(password.toCharArray()));
        }

        def.setName(name);
        def.setTitle((String) properties.get(ContentModel.PROP_TITLE));
        def.setDescription((String) properties.get(ContentModel.PROP_DESCRIPTION));

        if (nodeService.hasAspect(nodeRef, TransferModel.ASPECT_ENABLEABLE))
        {
            def.setEnabled((Boolean) properties.get(TransferModel.PROP_ENABLED));
        }
        else
        {
            // If the enableable aspect is not present then we don't want transfer failing.
            def.setEnabled(Boolean.TRUE);
        }
    }

    /* (non-Javadoc)
     * 
     * @see org.alfresco.service.cmr.transfer.TransferService#verify(org.alfresco.service.cmr.transfer.TransferTarget) */
    public void verify(TransferTarget target) throws TransferException
    {
        transmitter.verifyTarget(target);
    }

    /**
     * Utility to dump the contents of a file to the console
     * 
     * @param file
     *            File
     */
    private static void outputFile(File file) throws IOException
    {
        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), "UTF-8"));
        String s = reader.readLine();
        while (s != null)
        {
            System.out.println(s);
            s = reader.readLine();
        }
    }

    /**
     * Success transfer report
     */
    private NodeRef persistTransferReport(final String transferName, final Transfer transfer, final TransferTarget target, final TransferDefinition definition,
            final List<TransferEvent> events, final File snapshotFile, final Exception exception)
    {
        // persist the transfer report in its own transaction so it cannot be rolled back
        RetryingTransactionCallback<NodeRef> writeReportCallback = new RetryingTransactionCallback<NodeRef>() {
            @Override
            public NodeRef execute() throws Throwable
            {
                logger.debug("transfer report starting");
                NodeRef reportNode = null;
                if (exception != null)
                {
                    reportNode = transferReporter.createTransferReport(transferName, exception, target, definition, events, snapshotFile);
                }
                else
                {
                    reportNode = transferReporter.createTransferReport(transferName, transfer, target, definition, events, snapshotFile);
                }
                logger.debug("transfer report done");
                return reportNode;
            }
        };
        NodeRef reportNode = transactionService.getRetryingTransactionHelper().doInTransaction(writeReportCallback, false, true);
        return reportNode;
    }

    /**
     * Destination Transfer report
     * 
     * @return the node ref of the transfer report or null if there isn't one.
     */
    private NodeRef persistDestinationTransferReport(final String transferName,
            final Transfer transfer,
            final TransferTarget target)
    {
        // in its own transaction so it cannot be rolled back
        RetryingTransactionCallback<NodeRef> writeReportCallback = new RetryingTransactionCallback<NodeRef>() {
            @Override
            public NodeRef execute() throws Throwable
            {
                File tempDir = TempFileProvider.getLongLifeTempDir(FILE_DIRECTORY);
                File destReportFile = TempFileProvider.createTempFile("TRX-DREP", FILE_SUFFIX, tempDir);
                FileOutputStream destReportOutput = new FileOutputStream(destReportFile);
                transmitter.getTransferReport(transfer, destReportOutput);
                logger.debug("transfer report (destination) starting");

                NodeRef reportNode = transferReporter.writeDestinationReport(transferName, target, destReportFile);
                logger.debug("transfer report (destination) done");

                if (destReportFile != null)
                {
                    destReportFile.delete();
                }
                logger.debug("destination report temp file deleted");

                return reportNode;
            }
        };
        try
        {
            NodeRef reportNode = transactionService.getRetryingTransactionHelper().doInTransaction(writeReportCallback, false, true);
            return reportNode;
        }
        catch (Throwable e)
        {
            // there's nothing we can do here. - but we do not want the exception to propogate up.
            logger.debug("unexpected error while obtaining destination transfer report", e);
            return null;
        }
    }

    public void setTransferManifestNodeFactory(TransferManifestNodeFactory transferManifestNodeFactory)
    {
        this.transferManifestNodeFactory = transferManifestNodeFactory;
    }

    public void setActionService(ActionService actionService)
    {
        this.actionService = actionService;
    }

    public void setTransactionService(TransactionService transactionService)
    {
        this.transactionService = transactionService;
    }

    public void setTransferReporter(TransferReporter transferReporter)
    {
        this.transferReporter = transferReporter;
    }

    public void setCommitPollDelay(long commitPollDelay)
    {
        this.commitPollDelay = commitPollDelay;
    }

    public void setDescriptorService(DescriptorService descriptorService)
    {
        this.descriptorService = descriptorService;
    }

    public void setTransferVersionChecker(TransferVersionChecker transferVersionChecker)
    {
        this.transferVersionChecker = transferVersionChecker;
    }

    public void setNamespaceService(NamespaceService namespaceService)
    {
        this.namespaceService = namespaceService;
    }

    private class TransferStatus
    {
        boolean cancelMe = false;
        boolean cancelInProgress = false;
    }

}
